package com.walter.dynamic.thread.pool.core;

import com.walter.dynamic.thread.pool.ThreadPoolMetrics;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 *  自适应调整线程池
 * @author Walter
 * @date 2025/1/24 下午8:44
 **/
@Slf4j
public class AdaptiveThreadPoolAdjuster {
    private final DynamicThreadPoolExecutor threadPool;
    private final int minCorePoolSize;
    private final int maxCorePoolSize;

    private final double loadThresholdHigh = 0.75;
    // 负载高阈值
    private final double loadThresholdLow = 0.25;


    /**
     *  负载低阈值
     * @param threadPool 动态线程池
     * @param minCorePoolSize 最小核心线程数
     * @param maxCorePoolSize 最大核心线程数
     */
    public AdaptiveThreadPoolAdjuster(DynamicThreadPoolExecutor threadPool, int minCorePoolSize, int maxCorePoolSize) {
        this.threadPool = threadPool;
        this.minCorePoolSize = minCorePoolSize;
        this.maxCorePoolSize = maxCorePoolSize;
    }

    /**
     *  定期调整线程池大小
     * @param period 调整周期
     * @param unit 时间单位
     */
    public void startAdjustment(long period, TimeUnit unit) {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(this::adjust, period, period, unit);
    }

    /**
     *  调整线程池大小
     */
    private void adjust() {
        synchronized (this) { // 同步块，防止并发调整
            ThreadPoolMetrics metrics = threadPool.getMetrics();
            // 获取线程池的当前状态
            int currentCorePoolSize = threadPool.getExecutor().getCorePoolSize();
            if(log.isDebugEnabled()){
                log.debug("ThreadPoolMetrics>>activeThreads:{},poolSize:{},queueSize:{},totalTasks:{},completedTasks:{} | currentCorePoolSize:{}",
                        metrics.getActiveThreads(), metrics.getPoolSize(), metrics.getQueueSize(), metrics.getTotalTasks(), metrics.getCompletedTasks(),currentCorePoolSize);
            }
            // 活跃线程比例
            double load = (double) metrics.getActiveThreads() / currentCorePoolSize;
            // 根据负载情况调整核心线程数
            if (load > loadThresholdHigh && currentCorePoolSize < maxCorePoolSize) {
                // 负载过高，增加线程数
                int newCorePoolSize = Math.min(currentCorePoolSize + 2, maxCorePoolSize);
                // 调整核心线程数
                threadPool.adjustCorePoolSize(newCorePoolSize);
                // 最大线程数也相应调整，但不小于 newCorePoolSize
                threadPool.adjustMaxPoolSize(Math.max(newCorePoolSize, metrics.getMaxPoolSize()));
            } else if (load < loadThresholdLow && currentCorePoolSize > minCorePoolSize) {
                // 负载过低，减少线程数
                int newCorePoolSize = Math.max(currentCorePoolSize - 1, minCorePoolSize);
                // 调整核心线程数
                threadPool.adjustCorePoolSize(newCorePoolSize);
                // 最大线程数也相应调整，但不小于 newCorePoolSize
                threadPool.adjustMaxPoolSize(Math.max(newCorePoolSize, metrics.getMaxPoolSize()));
            }
        }
    }
}
